package com.example.mqtt.demo.pub.impl;

import com.example.mqtt.demo.config.MqttConfig;
import com.example.mqtt.demo.pub.IPubService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

/**
 * <B>概要说明：</B> <BR>
 *
 * @author zhoulin1@xsyxsc.com
 * @since 2021-07-21 16:24
 */
@Slf4j
@Component
public class PubServiceImpl extends MqttConfig implements MqttCallback, IPubService {

	private String brokerUrl = null;

	private final String colon = ":";
	private final String clientId = "mqtt_server_pub";

	private MqttClient mqttClient = null;
	private MqttConnectOptions connectionOptions = null;
	private MemoryPersistence persistence = null;

	/**
	 * Private default constructor
	 */
	private PubServiceImpl() {
		this.config();
		System.out.println("初始化生产者...");
	}

	/**
	 * Private constructor
	 */
	private PubServiceImpl(String broker, Integer port, Boolean ssl, Boolean withUserNamePass) {
		this.config(broker, port, ssl, withUserNamePass);
	}

	/**
	 * Factory method to get instance of MQTTPublisher
	 *
	 * @return MQTTPublisher
	 */
	public static PubServiceImpl getInstance() {
		return new PubServiceImpl();
	}

	/**
	 * Factory method to get instance of MQTTPublisher
	 *
	 * @param broker
	 * @param port
	 * @param ssl
	 * @param withUserNamePass
	 * @return MQTTPublisher
	 */
	public static PubServiceImpl getInstance(String broker, Integer port, Boolean ssl, Boolean withUserNamePass) {
		return new PubServiceImpl(broker, port, ssl, withUserNamePass);
	}

	@Override
	protected void config() {

		this.brokerUrl = this.TCP + this.broker + colon + this.port;
		this.persistence = new MemoryPersistence();
		this.connectionOptions = new MqttConnectOptions();
		try {
			this.mqttClient = new MqttClient(brokerUrl, clientId, persistence);
			this.connectionOptions.setCleanSession(false);
			this.connectionOptions.setConnectionTimeout(5);
			this.connectionOptions.setAutomaticReconnect(true);
			this.connectionOptions.setKeepAliveInterval(20);
			this.mqttClient.connect(this.connectionOptions);
			this.mqttClient.setCallback(this);

		} catch (MqttException me) {
			log.error("ERROR", me);
		}
	}

	@Override
	protected void config(String broker, Integer port, Boolean ssl, Boolean withUserNamePass) {

		String protocal = this.TCP;
		if (true == ssl) {
			protocal = this.SSL;
		}

		this.brokerUrl = protocal + this.broker + colon + port;
		this.persistence = new MemoryPersistence();
		this.connectionOptions = new MqttConnectOptions();

		try {
			this.mqttClient = new MqttClient(brokerUrl, clientId, persistence);
			this.connectionOptions.setCleanSession(false);
			this.connectionOptions.setConnectionTimeout(5);
			this.connectionOptions.setAutomaticReconnect(true);
			this.connectionOptions.setKeepAliveInterval(20);
			if (true == withUserNamePass) {
				if (password != null) {
					this.connectionOptions.setPassword(this.password.toCharArray());
				}
				if (userName != null) {
					this.connectionOptions.setUserName(this.userName);
				}
			}
			this.mqttClient.connect(this.connectionOptions);
			this.mqttClient.setCallback(this);
		} catch (MqttException me) {
			log.error("ERROR", me);
		}
	}

	@Override
	public void publishMessage(String topic, String message) {
		try {
			MqttMessage mqttmessage = new MqttMessage(message.getBytes());
			mqttmessage.setQos(this.qos);
			this.mqttClient.publish(topic, mqttmessage);
		} catch (MqttException me) {
			log.error("ERROR", me);
		}
	}

	/**
	 * This method is called when the connection to the server is lost.
	 *
	 * @param cause the reason behind the loss of connection.
	 */
	@Override
	public void connectionLost(Throwable cause) {
		log.info("Connection Lost");
	}

	/**
	 * Called when delivery for a message has been completed, and all
	 * acknowledgments have been received. For QoS 0 messages it is
	 * called once the message has been handed to the network for
	 * delivery. For QoS 1 it is called when PUBACK is received and
	 * for QoS 2 when PUBCOMP is received. The token will be the same
	 * token as that returned when the message was published.
	 *
	 * @param token the delivery token associated with the message.
	 */
	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		log.info("delivery completed");
	}

	/**
	 * This method is called when a message arrives from the server.
	 *
	 * <p>
	 * This method is invoked synchronously by the MQTT client. An
	 * acknowledgment is not sent back to the server until this
	 * method returns cleanly.</p>
	 * <p>
	 * If an implementation of this method throws an <code>Exception</code>, then the
	 * client will be shut down.  When the client is next re-connected, any QoS
	 * 1 or 2 messages will be redelivered by the server.</p>
	 * <p>
	 * Any additional messages which arrive while an
	 * implementation of this method is running, will build up in memory, and
	 * will then back up on the network.</p>
	 * <p>
	 * If an application needs to persist data, then it
	 * should ensure the data is persisted prior to returning from this method, as
	 * after returning from this method, the message is considered to have been
	 * delivered, and will not be reproducible.</p>
	 * <p>
	 * It is possible to send a new message within an implementation of this callback
	 * (for example, a response to this message), but the implementation must not
	 * disconnect the client, as it will be impossible to send an acknowledgment for
	 * the message being processed, and a deadlock will occur.</p>
	 *
	 * @param topic   name of the topic on the message was published to
	 * @param message the actual message.
	 * @throws Exception if a terminal error has occurred, and the client should be
	 *                   shut down.
	 */
	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		// Leave it blank for Publisher

	}

	@Override
	public void disconnect() {
		try {
			this.mqttClient.disconnect();
		} catch (MqttException me) {
			log.error("ERROR", me);
		}
	}

}
